package com.ermu.reactive.jdkaction;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

public class ReactiveApplication {
    public  static void main(String[] args){
        customerSubmissionPublisher();
    }
    private static void demoSubscribe(ReactivePublisher<Integer> publisher,String subscriberName){
        ReactiveSubscriber<Integer> subscriber  = new ReactiveSubscriber<>(4L,subscriberName);
        publisher.subscribe(subscriber);
    }
    private static void customerSubmissionPublisher(){
        ExecutorService executorService = ForkJoinPool.commonPool();
        try {
            ReactivePublisher<Integer> publisher  = new ReactivePublisher<>(executorService);
            demoSubscribe(publisher,  "One");
            demoSubscribe(publisher,  "Two");
            demoSubscribe(publisher,  "Three");
            IntStream.range(1,5).forEach(publisher::submit);
        }finally {
            executorService.shutdown();
            int shutdownDelaySec = 1;
            System.out.println("………………等待 " + shutdownDelaySec + " 秒后结束服务……………… ");
            try {
                executorService.awaitTermination(shutdownDelaySec, TimeUnit.SECONDS);
            } catch (InterruptedException ex) {
                System.out.println("捕获到 execService.awaitTermination()方法的异常: " + ex.getClass().getName());
            }finally {
                System.out.println("调用 execService.shutdownNow()结束服务...");
                List<Runnable> l = executorService.shutdownNow();
                System.out.println("还剩 "+l.size() + " 个任务等待被执行，服务已关闭 ");
            }
        }
    }
}
